/*
 * Copyright (c) China Telecom Cloud Technology Co., Ltd. 2024-2025. All rights reserved.
 */
#include <atomic>
#include <algorithm>
#include <csignal>
#include <cstring>
#include <sys/socket.h>
#include <netinet/in.h>
#include <unistd.h>
#include <cerrno>

#include <grpcpp/server_builder.h>

#include "KVCacheService.h"
#include "Segment.h"
#include "Logger.h"
#include "ConfigLoader.h"

#define MAX_SHUTTIME_TIME  100
#define MAX_WAITING_TIME   5
#define MAX_PORT_ATTEMPTS  5
#define MAX_VALID_PORT     65535
#define MIN_VALID_PORT     1
#define DEFAULT_PORT       50051


std::atomic<bool> shutdown_flag(false);


void SignalHandler(int /*signo*/) {
    shutdown_flag.store(true);
}



/**
 *! @brief 判断某个端口是否可用
 * 
 * @param port 
 * @return bool
 */
bool is_port_available(int port) {
    int sock = socket(AF_INET, SOCK_STREAM, 0);
    if (sock < 0) {
        LOG_ERROR_FMT("Socket creation failed: {}", strerror(errno));
        return false;
    }

    sockaddr_in addr{};
    addr.sin_family      = AF_INET;
    addr.sin_port        = htons(port);
    addr.sin_addr.s_addr = INADDR_ANY;

    int opt = 1;
    setsockopt(sock, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt));

    bool available = (bind(sock, 
                        reinterpret_cast<struct sockaddr  *>(&addr), 
                        sizeof(addr)
                    ) == 0);
    close(sock);

    return available;
}




/**
 *! @brief 初始化logger
 * 
 */
void initialize_logger() {
    const std::string LOG_PATH = ConfigLoader::get_config<std::string>(
                                    ConfigLoader::EnvVars::LOG_PATH, 
                                    DEFAULT_LOG_PATH
                                );
    Logger::instance().init(LOG_PATH + "KVClient_", 10 * 1024 * 1024);
}


/**
 *! @brief 设置信号处理器
 * 
 */
void setup_signal_handlers() {
    struct sigaction sa {};
    sa.sa_handler = SignalHandler;
    sigemptyset(&sa.sa_mask);
    sigaction(SIGINT, &sa, nullptr);
    sigaction(SIGTERM, &sa, nullptr);
}



/**
 *! @brief 解释配置文件中的端口设置并验证有效性
 * 
 * @return int 
 */
int parse_and_validate_port() {
    const std::string CLIENT_PORT_STR = ConfigLoader::get_config<std::string>(
                                            ConfigLoader::EnvVars::CLIENT_PORT, 
                                            DEFAULT_CLIENT_PORT
                                        );

    int port = DEFAULT_PORT;
    try {
        std::string port_str = (CLIENT_PORT_STR[0] == ':') 
                                ? CLIENT_PORT_STR.substr(1) 
                                : CLIENT_PORT_STR;
        port = std::stoi(port_str);
    } catch (const std::exception &e) {
        LOG_ERROR_FMT("Invalid port format: {}. Using "
                "default port 50051", CLIENT_PORT_STR);
        port = DEFAULT_PORT;
    }

    if (port < MIN_VALID_PORT || port > MAX_VALID_PORT) {
        LOG_ERROR_FMT("Invalid port number: {}. Using default port", port);
        port = DEFAULT_PORT;
    }
    return port;
}



/**
 *! @brief 启动gPRC服务端
 * 
 * @param port 
 * @param service 
 * @param server 
 * 
 * @return bool
 */
bool start_grpc_server(int port, 
                    KVCacheServiceImpl &service, 
                    std::unique_ptr<grpc::Server> &server) {
    int current_port    = port;
    int attempts        = 0;
    bool server_started = false;

    while (!server_started && attempts < MAX_PORT_ATTEMPTS) {
        // 设置gRPC绑定的地址与端口
        //! 这里好像使用localhost更安全
        std::string address = "0.0.0.0:" + std::to_string(current_port);


        // 如果端口不可用，则多次尝试
        if (!is_port_available(current_port)) {
            LOG_WARN_FMT("Port {} is unavailable: {}", 
                        current_port, 
                        strerror(errno)
                    );
            attempts++;
            if (attempts < MAX_PORT_ATTEMPTS) {
                LOG_WARN_FMT("Sleep 5s Waiting...");
                sleep(MAX_WAITING_TIME);
            }
            continue;
        }

        grpc::ServerBuilder builder;
        builder.AddListeningPort(address,                          // 监听IP与端口
                                grpc::InsecureServerCredentials()  // 允许非安全链接
                            );

        builder.AddChannelArgument("grpc.max_concurrent_streams",      // 最大并行流
                                KVC_GRPC_MAX_CONCURRENT_STREAMS
                            );
        builder.SetMaxReceiveMessageSize(KVC_GRPC_MAX_RECV_MSG_SIZE);  // 最大接收消息大小
        builder.SetMaxSendMessageSize(KVC_GRPC_MAX_SEND_MSG_SIZE);     // 最大发送消息大小

        //! 注册自定义服务对象
        // TODO(yangxianpku:2025.08.25): 不用改用static_cast
        builder.RegisterService(static_cast<KVCtrl::KVControlApi::Service *>(&service));
        builder.RegisterService(static_cast<kvcache::KVCacheService::Service *>(&service));


        // 启动服务
        server = builder.BuildAndStart();

        if (!server) {
            LOG_ERROR_FMT("Failed to start server on port {}", current_port);
            current_port++;
            attempts++;
        } else {
            server_started = true;
            LOG_INFO_FMT("Server started successfully on port: {}", current_port);
            service.init();
        }
    }
    return server_started;
}



/**
 *! @brief 循环判断停止信号
 * 
 * @param server 
 */
void wait_for_shutdown(grpc::Server *server) {
    while (!shutdown_flag.load()) {
        std::this_thread::sleep_for(
            std::chrono::milliseconds(MAX_SHUTTIME_TIME));
    }

    LOG_INFO_FMT("Shutting down server...");
    server->Shutdown();
    LOG_INFO_FMT("Server shutdown complete");
}



int main()
{
    initialize_logger();
    setup_signal_handlers();
    const int port = parse_and_validate_port();

    KVCacheServiceImpl service;
    std::unique_ptr<grpc::Server> server;


    if (!start_grpc_server(port, service, server)) {
        LOG_ERROR_FMT("Failed to start server after {} "
                "attempts. Exiting.", MAX_PORT_ATTEMPTS);
        return EXIT_FAILURE;
    }

    //! 这里后续还有wait_for_shutdown逻辑，并没有调用server->Wait();阻塞线程
    wait_for_shutdown(server.get());
    return EXIT_SUCCESS;
}